package com.autonavi.yunda.yunji.core.engine;

import com.autonavi.yunda.yunji.common.exception.AmapException;
import com.autonavi.yunda.yunji.common.exception.AmapExceptionCode;
import com.autonavi.yunda.yunji.common.thread.TraceableThreadPoolTaskExecutor;
import com.autonavi.yunda.yunji.common.utils.JsonUtils;
import com.autonavi.yunda.yunji.core.engine.context.EngineContext;
import com.autonavi.yunda.yunji.core.engine.context.NodeContext;
import com.autonavi.yunda.yunji.core.enums.AccessStatus;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static com.autonavi.yunda.yunji.core.config.EngineConstant.PARAM_VALIDATE_FUN_NAME;

/**
 * @author cizhong.wcs
 * @date 2022/5/31 11:50
 * 并行执行引擎
 */
@Slf4j
@Component
public class ParallelEngineWorker extends AbstractEngineWorker {
    private static final Logger engineLogger = LoggerFactory.getLogger("logger.engine.exec.log");
    @Autowired(required = false)
    TraceableThreadPoolTaskExecutor engineWorkerExecutor;

    public void doExe(EngineContext engineContext, Map<String, Object> inputParams) {
        engineContext.startTime = System.currentTimeMillis();
        try {
            engineContext.injectParams(inputParams);
            engineContext.invoke(PARAM_VALIDATE_FUN_NAME);
            exeSync(engineContext.getStartContext(), AccessStatus.ACCESSIBLE);
            try {
                for (; ; ) {
                    TimeUnit.NANOSECONDS.sleep(100);
                    engineContext.checkTimeout();
                    if (engineContext.finishedStatus != 0) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                throw new AmapException(AmapExceptionCode.COMMON_SERVER_ERROR, "异常中断");
            }
        } finally {
            engineContext.endTime = System.currentTimeMillis();
            engineContext.executeTime = engineContext.endTime - engineContext.startTime;
            if (engineContext.debug) {
                engineLogger.info("exe end engineContext :{}", JsonUtils.toString(engineContext));
            }
        }
    }

    private void exeAsync(NodeContext context, AccessStatus accessible) {
        if (context.failCheck()) {
            return;
        }
        Future<?> future = engineWorkerExecutor.submit(() -> exeSync(context, accessible));
        context.setExecuteFuture(future);
    }

    private void exeSync(NodeContext context, AccessStatus accessible) {
        engineLogger.info("node {} exe start", context.getName());
        checkAndExecNode(context, accessible);
        triggerNextNodes(context);
        context.setEndTime(System.currentTimeMillis());
        context.setExecuteTime(context.getEndTime() - context.getStartTime());
        engineLogger.info("node {} exe end", context.getName());
    }

    /**
     * 判断条件并触发下游节点执行
     */
    private void triggerNextNodes(NodeContext context) {
        List<Pair<NodeContext, AccessStatus>> pairs = fetchNextNodes(context);
        if (context.failCheck()) {
            return;
        }
        //优化:如果下游只有一个节点,则不创建新线程异步执行.
        if (pairs.size() == 1) {
            NodeContext nextNodeContext = pairs.get(0).getKey();
            AccessStatus accessStatus = pairs.get(0).getValue();
            this.exeSync(nextNodeContext, accessStatus);
            //若同步执行,则直接继承上游节点的future
            if (context.getExecuteFuture() != null) {
                nextNodeContext.setExecuteFuture(context.executeFuture);
            }
        } else {
            pairs.forEach(pair -> this.exeAsync(pair.getKey(), pair.getValue()));
        }
    }
}
